package communication.common.business.base;

import com.hazelcast.config.Config;
import io.vertx.core.*;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.EventBusOptions;
import io.vertx.core.spi.cluster.ClusterManager;
import io.vertx.spi.cluster.hazelcast.ConfigUtil;
import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;
import lombok.extern.slf4j.Slf4j;

/**
 *  eventBus管理器
 * @author mac
 */
@Slf4j
public class EventBusManager {

    private EventBus eventBus;


    /**
     * 初始eventBus为本地的eventBus
     */
    public void initLocalEventBus(Vertx vertx){
        this.eventBus = vertx.eventBus();
    }


    /**
     * 使用Hazelcast初始化一个集群Eventbus，需要配Hazelcast的配置文件
     */
    public void initClusterEventBus(){
        // 加载cluster.xml中的配置，将其转换为了Config对象
        Config hazelcastConfig = ConfigUtil.loadConfig();
        ClusterManager mgr=new HazelcastClusterManager(hazelcastConfig);
        VertxOptions options = new VertxOptions().setClusterManager(mgr);
        Vertx.clusteredVertx(options,res->{
                if(res.succeeded()){
                    log.info("cluster event bus init success ! ");
                    Vertx vertx=res.result();
                    this.eventBus=vertx.eventBus();
                    // 注册一个consumer,用来打印消息
                    this.eventBus.consumer("fcr.server",message -> {
                        log.info("server-eventBus received:{}",message.body());
                    });
                }else{
                    log.error("cluster eventbus init fail ! error:{} ",res.cause().getMessage());
                }
            });
    }

    /**
     * 获取eventBus为集群的eventBus
     * @param host 远程集群的主机地址
     * @param port 远程集群的主机端口
     */
    public void getClusterEventBus(String host, int port ){

        VertxOptions vertxOptions=new VertxOptions()
                .setEventBusOptions(new EventBusOptions()
                .setClusterPublicHost(host)
                .setClusterPublicPort(port));

        Vertx.clusteredVertx(vertxOptions,res->{
           if(res.succeeded()){
                log.info("cluster eventbus get success ! host:{},port{}",host,port);
                Vertx vertx=res.result();
                this.eventBus=vertx.eventBus();
           }else{
               log.error("cluster eventbus get fail ! host:{},port{},error:{} ",host,port,res.cause().getMessage());
           }
        });
    }

    /**
     * 获取集群的cluster的future版本
     * @param host 远程集群的主机地址
     * @param port 远程集群的主机端口
     */
    public Promise<Boolean> getClusterEventBusFuture(String host, int port ){
        Promise<Boolean> promise=Promise.promise();
        VertxOptions vertxOptions=new VertxOptions()
                .setEventBusOptions(new EventBusOptions()
                        .setClusterPublicHost(host)
                        .setClusterPublicPort(port));

        Vertx.clusteredVertx(vertxOptions,res->{
            if(res.succeeded()){
                log.info("cluster eventbus get success ! host:{},port{}",host,port);
                Vertx vertx=res.result();
                this.eventBus=vertx.eventBus();
                promise.complete(true);
            }else{
                log.error("cluster eventbus get fail ! host:{},port{},error:{} ",host,port,res.cause().getMessage());
                promise.complete(false);
            }
        });

        return promise;

    }


    /**
     * 通过eventBus向address发送消息msg (如果同一个address有多个consumer，则会随机发送到某一个consumer上)
     * @param address 消息要发送到的地址
     * @param msg 要发送的消息
     * @param options  消息头（可以为null）
     */
    public void sendMsg(String address,Object msg, DeliveryOptions options) throws Exception {
        //集群模式下的eventBus创建是异步的， 直接使用前需要先判断是否创建成功
        if(this.eventBus==null){
            throw new Exception("event bus is null, can not send msg !");
        }

        if(options!=null){
            this.eventBus.send(address,msg,options);
        }else{
            this.eventBus.send(address,msg);
        }

    }

    /**
     * 通过eventBus向所有address发送消息msg
     * @param msg 要发送的消息
     */
    public void publishMsg(String address, Object msg ) throws Exception {
        //集群模式下的eventBus创建是异步的， 直接使用前需要先判断是否创建成功
        if(this.eventBus==null){
            throw new Exception("event bus is null, can not send msg !");
        }

        this.eventBus.publish(address,msg);
    }

}
